SLAs & Callbacks
Reacting to State Changes, Not Just Final Success or Failure
Every notification integration covered in Module 09 needs a trigger point - something in Airflow that says "call this code now." Callbacks are that trigger point: functions Airflow invokes automatically on specific task/DAG state transitions, without needing an extra task in the DAG to check for them.
The Callback Types
| Callback | Fires When |
|---|---|
on_success_callback |
A task instance succeeds |
on_failure_callback |
A task instance fails (after exhausting retries) |
on_retry_callback |
A task instance is about to retry |
on_execute_callback |
A task instance starts executing |
sla_miss_callback |
A DAG's tasks haven't finished within its defined SLA window |
Task-Level and DAG-Level Callbacks
def notify_failure(context):
task_id = context["task_instance"].task_id
dag_id = context["dag"].dag_id
print(f"ALERT: {dag_id}.{task_id} failed")
# In practice: call the PagerDuty/Slack/Teams integration from Module 09 here
@dag(
schedule="@daily",
start_date=...,
on_failure_callback=notify_failure, # DAG-level: fires if the DAG run itself fails
)
def sales_pipeline():
@task(on_failure_callback=notify_failure) # task-level: fires for this specific task
def extract():
...
A DAG-level callback fires once per DAG run failure; a task-level callback fires per task. Most alerting setups use task-level callbacks on the handful of genuinely critical tasks, plus one DAG-level callback as a catch-all.
SLAs — Time-Based Expectations
An SLA declares "this task should finish within X of the DAG run's start" — independent of whether it ultimately succeeds or fails:
from datetime import timedelta
@task(sla=timedelta(hours=2))
def extract_from_slow_api():
...
If this task is still running two hours after the DAG run started, Airflow fires sla_miss_callback (configured at the DAG level) — a signal that something is abnormally slow, which is a different, earlier warning than waiting for outright failure.
def handle_sla_miss(dag, task_list, blocking_task_list, slas, blocking_tis):
print(f"SLA missed for: {task_list}")
# Alert the team that something is running unusually slowly
@dag(schedule="@daily", start_date=..., sla_miss_callback=handle_sla_miss)
def sales_pipeline():
...
A task queued for 90 minutes behind other work, then running for 10 minutes, still counts as 100 minutes against its SLA - the clock starts at the DAG run's scheduled time, not whenever that specific task actually began executing. A queue backlog can trigger SLA misses even for tasks that ran quickly once they started.
Anything that must run regardless of whether the DAG ultimately succeeds or fails (a cleanup step, a "pipeline finished" notification either way) is usually better modeled as a task with
trigger_rule="all_done" (covered in the TaskGroups/Branching material) than as a callback - callbacks are best reserved for genuinely reactive, out-of-band signals like alerting, not steps that are really part of the pipeline's own logic.